package com.bbxdemo.kafka.produce;


import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Properties;
import java.util.concurrent.ExecutionException;


/**
 * 发送消息
 */
@Slf4j
@RestController
@RequestMapping("/produce/SendMsg")
public class SendMsg {





    /**
     * 异步发送消息
     * @param msg
     */
    @RequestMapping("/send")
    public void send(String msg){
        log.info("msg[{}]", msg);
        //设置生产者属性
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "master11.bbx.com:9092");
        //设置序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        //创建发送的内容
        ProducerRecord producerRecord = new ProducerRecord("bbx-first-topic",msg+"--"+System.currentTimeMillis());
        //发送数据
        producer.send(producerRecord);
        //关闭资源
        producer.close();

    }



    /**
     * 发送消息  并且回调
     * @param msg
     */
    @RequestMapping("/sendAndCall")
    public void sendAndCall(String msg){
        log.info("msg[{}]", msg);
        //设置生产者属性
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "master11.bbx.com:9092");
        //设置序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        //创建发送的内容
        ProducerRecord producerRecord = new ProducerRecord("bbx-first-topic",msg+"--"+System.currentTimeMillis());
        //发送数据
        producer.send(producerRecord,new sendMsgCallback());
        //关闭资源
        producer.close();

    }






    /**
     * 同步发送消息
     * @param msg
     */
    @RequestMapping("/sendSync")
    public void sendSync(String msg) throws ExecutionException, InterruptedException {
        log.info("msg[{}]", msg);
        //设置生产者属性
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "master11.bbx.com:9092");
        //设置序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        //创建发送的内容
        ProducerRecord producerRecord = new ProducerRecord("bbx-first-topic",msg+"--"+System.currentTimeMillis());
        //发送数据
        RecordMetadata o = (RecordMetadata)producer.send(producerRecord).get();
        log.info("sendSync result topic[{}]", o.topic());
        //关闭资源
        producer.close();

    }



}

/**
 * 回调类
 */
@Slf4j
class sendMsgCallback implements Callback{

    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if(exception == null){
            log.info("topic[{}],partition[{}]", metadata.topic(), metadata.partition());
        }else{
            log.error("发送消息出现错误[{}]", exception);
        }
    }
}
